
/*
 * Spread implementation of JGCS - Group Communication Service.
 * Copyright (C) 2004,2006 Jose' Orlando Pereira, Universidade do Minho
 *
 * jop@di.uminho.pt - http://gsd.di.uminho.pt/~jop
 * jgcs@lasige.di.fc.ul.pt
 *
 * Departamento de Informatica, Universidade do Minho
 * Campus de Gualtar, 4710-057 Braga, Portugal
 *
 * See COPYING for licensing details.
 */

package net.sf.jgcs.spread;

import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Logger;

import net.sf.jgcs.AbstractControlSession;
import net.sf.jgcs.AbstractPollingProtocol;
import net.sf.jgcs.ControlSession;
import net.sf.jgcs.DataSession;
import net.sf.jgcs.GroupConfiguration;
import net.sf.jgcs.JGCSException;
import net.sf.jgcs.spread.jni.Mailbox;


public class SpProtocol extends AbstractPollingProtocol {
	
	private static int BUFFER_INITIAL_SIZE = 15360;
	
	public static int getBUFFER_INITIAL_SIZE() {
		return BUFFER_INITIAL_SIZE;
	}

	public static void setBUFFER_INITIAL_SIZE(int buffer_initial_size) {
		BUFFER_INITIAL_SIZE = buffer_initial_size;
	}


	private Mailbox mb;
	private SpDataSession defaultData;

	private String daemonAddress;
	private String processName;

	private AtomicBoolean connected = new AtomicBoolean(false);
	
	private static Logger logger = Logger.getLogger(SpProtocol.class.getName());

	SpProtocol(Mailbox mb, String daemonAddress, String processName) throws JGCSException {
		this.mb = mb;
		if (processName==null)
			processName=""+System.nanoTime()/1000%100000000;

		this.daemonAddress = daemonAddress;
		this.processName = processName;
	}

	protected synchronized void connect() throws SpException {
		if (connected.get() == false) {
			logger.fine("Connecting to spread daemon [daemon_address="
					+ daemonAddress + ", process_name=" + processName + "]");
			int ret=mb.C_connect(daemonAddress, processName, false, true);
			
			if (ret<0) throw new SpException(ret, null);
			
			connected.set(true);
			start();
		}
	}

	private synchronized void createSessions(SpGroup group) {
		defaultData=new SpDataSession(mb, this, group);
		putSessions(group, new SpControlSession(mb, this, group), defaultData);
	}

	protected synchronized void removeSessions(GroupConfiguration group) {
		SpControlSession control=(SpControlSession)lookupControlSession(group);
		control.handleClose();
		super.removeSessions(group);
	}

	public DataSession openDataSession(GroupConfiguration group) throws JGCSException {
		DataSession data=lookupDataSession(group);
		if (data==null) {
			createSessions((SpGroup)group);
			data=lookupDataSession(group);
		}
		return data;
	}

	public ControlSession openControlSession(GroupConfiguration group) throws JGCSException {
		ControlSession control=lookupControlSession(group);
		if (control==null) {
			createSessions((SpGroup)group);
			control=lookupControlSession(group);
		}
		return control;
	}

	protected void read() {		
		Mailbox.ReceiveArgs info=new Mailbox.ReceiveArgs();
		int allocate_size = BUFFER_INITIAL_SIZE, ret = -1;
		ByteBuffer mess=null;

		mess = ByteBuffer.allocate(allocate_size);
		ret = mb.C_receive(info, mess);
		
		if (ret == SpException.BUFFER_TOO_SHORT && info.endian_mismatch < 0) {
			allocate_size = -info.endian_mismatch;
				logger.info("Spread returned BUFFER_TOO_SHORT. " +
						"Increasing buffer to required size: " + allocate_size);
			mess = ByteBuffer.allocate(allocate_size);
			ret = mb.C_receive(info, mess);
		}
		
		// error
		if (ret < 0) {
			logger.fine("an error ocurred while receiving message: " + ret);
			if (ret == SpException.CONNECTION_CLOSED
					|| ret == SpException.ILLEGAL_SESSION
					|| ret == SpException.NET_ERROR_ON_SESSION) {
				connectionClosedBySpread();
				
			} else {
				GroupConfiguration group = new SpGroup(info.sender);
				logger.fine("GROUP NULL???"+(group == null));
				SpDataSession data = (SpDataSession) lookupDataSession(group);
				if (data == null) {
					logger.info("no group associated with the message");
					data = defaultData;
				}
				else{
					logger.fine("an error ocurred while receiving message: ");
				}
				data.notifyExceptionListeners(new SpException(ret, null));
			}
			return;
		}
		

		mess.flip();
		if ((info.service_type & SpService.REGULAR_MESS) == 0) {
			GroupConfiguration group = new SpGroup(info.sender);
			AbstractControlSession control = lookupControlSession(group);
			if (control==null)
				return;
			((SpControlSession) control).deliverView(info, mess);
		} else {
			// If this is a subcast, info.groups contains all targets
			// and then the enclosing group name
			String name = info.groups[info.groups.length-1];
			GroupConfiguration group = new SpGroup(name);
			DataSession data = lookupDataSession(group);
			// If we don't know the group, it is directed at the process itself
			if (data==null)
				data=defaultData;
			((SpDataSession) data).deliverMessage(info,mess);
		}
	}
	

	protected void connectionClosedBySpread() {
		if (connected.getAndSet(false)) {
			
			Collection<AbstractControlSession> sessions = null;
			synchronized (this) {
				sessions = this.controlSessions.values();
			}
			
			logger.fine("Disconnected from daemon. Stopping protocol reader");
			this.stop();

			// notify all groups connected to this daemon
			for (AbstractControlSession c : sessions) {
				((SpControlSession) c).notifyMemberRemoved();
			}
		}
	}
}